package edu.gmu.middleware;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * TCPChannelSink reads/writes bytes using the TCP protocol.
 *
 * @author  James H. Pope
 * @version $Revision:$ $Date:$
 */
public class TCPStreamSink implements Sink
{
    /**
     * Default buffer sizes for UDP Datagrams is 8192 - theoretical
     * is 65,507 (IPv4) or 65536 (IPv6)
     */
    //public static final int DEFAULT_SO_BUFFER_SIZE = 8192;
    
    //----------------------------- Sink Variables ---------------------------//
    private Socket socket = null;
    
    private SocketAddress remoteAddress = null;
    private SocketAddress localAddress = null;
    private int readBufferSize = 0;
    
    private volatile boolean opened = false;
    private volatile boolean closed = false;
    
    private OutputStream out = null;
    private InputStream  in  = null;
    
    private Lock readLock  = new ReentrantLock(true); // make fair
    private Lock writeLock = new ReentrantLock(true); // make fair
    
    /**
     * Creates a new Client TCPStreamSink with specified values.
     * @param remoteAddress
     * @param readBufferSize
     */
    public TCPStreamSink( SocketAddress remoteAddress, int readBufferSize )
    {
        this( remoteAddress, null, readBufferSize );
    }
    
    /**
     * Creates a new Client TCPStreamSink with specified values.
     * @param remoteAddress
     * @param localAddress
     * @param readBufferSize
     */
    public TCPStreamSink(   SocketAddress remoteAddress,
                            SocketAddress localAddress,
                            int readBufferSize )
    {
        this.remoteAddress  = remoteAddress;
        this.localAddress   = localAddress;
        this.readBufferSize = readBufferSize;
    }
    
    /**
     * Creates a new Client TCPStreamSink with specified values.
     * @param clientSocket
     * @param readBufferSize
     */
    public TCPStreamSink( Socket clientSocket, int readBufferSize )
    {
        this.socket        = clientSocket;
        this.remoteAddress = socket.getRemoteSocketAddress();
        this.localAddress  = socket.getLocalSocketAddress();
        this.readBufferSize = readBufferSize;
    }
    
    /**
     * Determones if Sink is open for reading/writing.
     * @return true if connected, false otherwise
     */
    public boolean isConnected()
    {
        return (this.closed == false && this.opened == true);
    }
    
    /**
     * Opens the Sink for reading from or writing to.
     */
    public void connect() throws IOException
    {
        // Check State and complain if not good
        if( this.opened == true )
        {
            throw new IllegalStateException( "Sink already opened " + 
                    "- closed=" + closed + " opened=" + opened );
        }
        
        /*
         * Two states when this method is called
         * 1. We already have socket (e.g. from ServerSocket.accept) - blocks
         * 2. We do not have socket
         */
        if( this.socket == null )
        {
            // "Connect" to remoteAddress - blocks until connected
            this.socket = new Socket( );
            if( this.localAddress != null )
            {
                this.socket.bind(this.localAddress);
            }
            this.socket.connect(this.remoteAddress);
        }
        else
        {
            // nothing to do but get streams
        }
        
        this.out = new BufferedOutputStream( this.socket.getOutputStream() );
        this.in  = new BufferedInputStream( this.socket.getInputStream() );
        //this.out = this.socket.getOutputStream();
        //this.in  = this.socket.getInputStream();
        
        //this.socket.setReceiveBufferSize(bufferSize);
        //this.socket.setSendBufferSize(bufferSize);
        
        
        // Open before starting threads - otherwise race condition
        this.opened = true;
        
        /*
         * I maintain that only one readerThread is necessary and thus
         * reduces the overheader (i.e. one inBuffer) - however, should
         * it be found that listeners are taking too long in callbacks
         * then we might consider adding another readThread - made easy.
         */
        this.readerThreads = new LinkedList();
        int numReaderThreads = 1;
        for (int i = 0; i < numReaderThreads; i++)
        {
            ByteBuffer inBuffer = ByteBuffer.allocate(readBufferSize);
            Thread readerThread = this.makeThread(  "TCPStreamSink-"+
                                                    this.localAddress+
                                                    "-"+i, inBuffer );
            this.readerThreads.add( readerThread );
            readerThread.start();
        }
    }
    
    /**
     * Override ancestor.  Close Socket.  Called from Close when the
     * Connection.close() method is invoked.  Subclasses should close
     * their Socket.
     */
    public void disconnect()
    {
        this.closed = true;
        this.running = false;
        
        // Typically should not be null - possinly if creation failed.
        if( this.socket != null )
        {
            try
            {
                this.socket.close();
                this.out    = null;
                this.in     = null;
                this.socket = null;
            }
            catch(IOException ioe)
            {
                // ignore
            }
        }
        
        // We can get two threads in calling this at same time
        // For now this is not a problem - should you null readerThreads 
        // it becomes a problem while one is iterating the other nulls it!
        if( this.readerThreads != null )
        {
            for (int i = 0; i < readerThreads.size(); i++)
            {
                Thread readerThread = (Thread)readerThreads.get(i);
                readerThread.interrupt();
            }
            //this.readerThreads = null;
        }
    }
    
    
    //----------------------------- Sink Methods -----------------------------//
    
    /**
     * Reads from the Sink (usually a network Socket) into buffer.
     * This method blocks until data is ready.
     * @param in
     * @return number of bytes headerOffset
     */
    public Header read( ByteBuffer in ) throws IOException
    {   
        this.readLock.lock();
        try
        {
            // Check State and complain if not good
            if( this.isConnected() == false )
            {
                throw new IOException( "Sink not in state to send " + 
                        "- closed=" + closed + " opened=" + opened );
            }

            //if( this.socket == null || this.socket.isInputShutdown() )
            //{
            //    throw new IllegalStateException( "Socket invalid " + 
            //            ((socket==null)?"null": "inputShutdown")  );
            //}

            /*
             * Block until we have what we think is the header
             */
            in.clear();
            byte[] inBytes = in.array();
            // Make very sure to check result - does not throw exception if -1!!!!
            // This assumes we can headerOffset Header.SIZEOF in one headerOffset???  Loop?
            int headerOffset = Header.SIZEOF;
            readFully( this.in, inBytes, 0, headerOffset );
            //if( headerOffset != Header.SIZEOF )
            //{
            //    String s = "Invalid return socket.inputStream.read: "+headerOffset;
            //    throw new IOException(s);
            //}

            //print( in, 12 );

            in.limit( Header.SIZEOF );
            Header header = new Header();
            header.unmarshal(in);


            if( header.getEndianFlag() == null )
            {
                String s = "Unknown header " + Header.toString(in);
                throw new IOException( s );
            }

            /*
             * Now we have enough to headerOffset rest of message
             */
            int length         = header.getLength();
            //System.out.println("Remaining: " + remaining + " Length: " + length);
            int bodyOffset = headerOffset;
            readFully( this.in, inBytes, bodyOffset, length );

            /*
             * Note - in is ready to headerOffset from position Header.SIZEOF
             * However, the unmarshaler expects to have it positioned at
             * the first index (i.e. reading the header).  Thus we end up
             * unmarshaling the header twice - here and the unmarshaler.
             */
            in.limit( length );
            in.position(0);
            //in.flip();
            //System.out.println( "Reading - length="+length );
            //System.out.println( "Length: " + length +" Limit: " + in.limit() );  
            return header; // believe it - this works with finally
        }
        finally
        {
            this.readLock.unlock();
        }
    }
    
    
    /**
     * Utility to readFully - similar to DataInputStream
     * @param in
     * @param b
     * @param offset
     * @param length
     * @throws java.io.IOException
     */
    private static void readFully(  InputStream in, byte[] b,
                                    int offset, int length ) throws IOException
    {
        while( offset < length )
        {
            int bytesRead = 0;
            try
            {
                bytesRead = in.read(b, offset, length-offset);
                
                if( bytesRead < 0 )
                {
                    String s = "Invalid return length: "+length
                            + " offset=" + offset + " bytesRead="+bytesRead;
                    throw new EOFException(s);
                }
                offset += bytesRead;
            }
            catch(IndexOutOfBoundsException e)
            {
                String s = "IndexOutOfBoundsException: offset=" + offset +
                        " bytesRead="+bytesRead + " length="+length;
                throw new IOException( s );
            }
            
        }
    }
    
    /**
     * Writes the contents of buffer to the Sink.  Though not as
     * significant as a headerOffset, this call is also blocking.
     * @param out
     * @return number of bytes sent
     */
    public int write( ByteBuffer out ) throws IOException
    {  
        this.writeLock.lock();
        try
        {
            // Check State and complain if not good
            if( this.isConnected() == false )
            {
                throw new IOException( "Sink not in state to send " + 
                        "- closed=" + closed + " opened=" + opened );
            }

            /*
             * Note that we do not support directAllocated byteBuffers.
             */
            out.flip();
            byte[] bytes = out.array();
            int limit    = out.limit();

            //System.out.println( "Writing - limit="+limit );
            // This should block until all bytes written
            this.out.write( bytes, 0, limit );
            this.out.flush();
            //System.out.println( "Sent message of size " + limit );

            // Do not clear before returning - router sends to multiple
            // using same!!
            // If all went well - return number of bytes written
            return limit; // believe it - this works with finally
        }
        finally
        {
            this.writeLock.unlock();
        }
    }
    
    
    
    //---------------------------- Event Methods -----------------------------//
    //
    //------------------------------------------------------------------------//
    
    private volatile boolean running    = false;
    private List sinkListeners          = new ArrayList();
    private List readerThreads          = null;
    
    
    private Thread makeThread( String name, final ByteBuffer in )
    {
        Runnable runnable = new Runnable()
        {
            public void run()
            {
                System.out.println("Buffer capacity " + in.capacity());
                running = true;
                while( running == true )
                {
                    
                    try
                    {
                        // Critical design decision concerning in buffer
                        // Each reader thread has it's own
                        Header header = read( in );
                        //System.out.println("SinkManager Received " + packet);
                        if( running == true )
                        {
                            // Callbacks (e.g. Dispatcher) may take some time
                            
                            /*
                             * Notifies listeners interested in serialized
                             * messages - e.g. recorders and routers.
                             */
                            fireDataReceived(header, in);
                        }
                    }
                    catch( Exception e )
                    {
                        //System.out.println("WE GOT ABNORMAL !!!!!!!!!");
                        // If our stop has been called - exit thread
                        if( running == true )
                        {
                            /*
                             * Catch io and user callback exceptions
                             * Note: Fairly dubious to allow clients
                             * to properly handle exceptions???
                             */
                            fireDisconnected(in,e);
                        }
                        // else ignore - we are stopping
                    }
                }
                // Should we stop running - close the sink
                running = false; // by definition, we are close enough to done
                disconnect();
                System.out.println("###############################");
                System.out.println("# Reader is exitting - bye!");
                System.out.println("###############################");
            }
        };
        // Overwrites any previous thread - are we reentrant?
        Thread t = new Thread(runnable);
        //thread.setDaemon(true);
        t.setName( name );
        
        return t;
    }
    
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param in
     * @param e
     */
    public void fireDisconnected( ByteBuffer in, Exception e )
    {
        //Do this in order to only create Event once
        SinkDisconnectedEvent event = null;
        
        // Determine type of event based on state
        int type = SinkDisconnectedEvent.NORMAL;
        if( this.isConnected() == true )
        {
            type = SinkDisconnectedEvent.ABNORMAL;
        }

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.sinkListeners.size(); i++)
        {
            // lazy initialization
            if(event == null)
            {
                event = new SinkDisconnectedEvent(  this,
                                            (long)System.currentTimeMillis(),
                                            type,
                                            in,
                                            this,
                                            e
                                     );
            }
            // Get listener
            SinkListener listener = ((SinkListener)this.sinkListeners.get(i));
            // Call listener
            listener.sinkDisconnected( event );
        }
    }
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param header
     * @param in
     */
    public void fireDataReceived( Header header, ByteBuffer in )
    {
        //Do this in order to only create Event once
        SinkDataEvent event = null;

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.sinkListeners.size(); i++)
        {
            // lazy initialization
            if(event == null)
            {
                event = new SinkDataEvent(  this,
                                        (long)System.currentTimeMillis(),
                                        header,
                                        in,
                                        this
                                     );
            }
            // Get listener
            SinkListener listener = ((SinkListener)this.sinkListeners.get(i));
            // Call listener
            listener.dataReceived( event );
        }
    }
    
    /**
     * Adds to list of listeners. First checks to see if
     * listener is already in list, if it is it does not add it a second time.
     * @param listener to add.
     */
    public void addSinkListener( SinkListener listener )
    {
        if( !this.sinkListeners.contains(listener) )
        {
            this.sinkListeners.add(listener);
        }
    }

    /**
     * Removes from list of listeners.
     * @param listener to remove.
     */
    public void removeSinkListener( SinkListener listener )
    {
        this.sinkListeners.remove(listener);
    }
        
        
    /**
     * Solely to be used for debugging.
     * @param inBuffer
     * @param length
     */
    private static void print( ByteBuffer inBuffer, int length )
    {
        byte[] in = inBuffer.array();
        System.out.print( "position="+inBuffer.position() + " limit="+inBuffer.limit() );
        for (int j = 0; j < length; j++)
        {
            String s = Integer.toHexString( in[j]&0xff );
            if( s.length() == 1 ) s = "0"+s;
            System.out.print( " " + s );
        }
        System.out.println( "" );
    }
}

